# -*- coding: utf-8 -*-
#
# Copyright 2007 - Celso Pinto <cpinto@7syntax.com>
#
# Licensed under the Apache License, Version 2.0 (the "License"); 
# you may not use this file except in compliance with the License. 
# You may obtain a copy of the License at 
#
#      http://www.apache.org/licenses/LICENSE-2.0 
#
# Unless required by applicable law or agreed to in writing, software 
# distributed under the License is distributed on an "AS IS" BASIS, 
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
# See the License for the specific language governing permissions and 
# limitations under the License.
#

#
# TODO implement dynamic thread pool for ConcurrentProcessor
#      currently the max_pool_size will be used for thread creation
#

import Queue
from threading import Lock

from exceptions import InvalidSettings,RollbackException
from generic import DaemonThread

__all__ = ['BaseProcessor','ConcurrentProcessor','SerializedProcessor']

class BaseProcessor(object):
    def __init__(self):
        pass
    
    def invoke_transacted(self,m):
        ctx = m.context
        if ctx.in_transaction:
            ctx.transaction_id = m.conn.begin()

        try:
            if not m.config.auto_acknowledge:
                m.acknowledge()

            m.config.worker(m)

            if ctx.in_transaction and ctx.rollback:
                raise RollbackException()

        except Exception,e:
            if ctx.in_transaction:
                m.conn.abort(transaction=ctx.transaction_id)
                
            if not isinstance(e,RollbackException):
                raise e
            
        else:
            if ctx.in_transaction:
                m.conn.commit(transaction=ctx.transaction_id)

    def invoke_simple(self,m):
        m.config.worker(m)
        if not m.config.auto_acknowledge:
            m.acknowledge()
    
    def invoke(self,m):
        try:
            if m.context.in_transaction:
                self.invoke_transacted(m)
            else:
                self.invoke_simple(m)
        except Exception,e:
            print e
            
    def process(self,message):
        pass
    
    def stop(self):
        pass

class SerializedProcessor(BaseProcessor):
    def __init__(self):
        super(SerializedProcessor,self).__init__()
    def process(self,message):
        self.invoke(m)

class ConcurrentProcessor(BaseProcessor):
    def __init__(self,min_size,max_size):
        super(ConcurrentProcessor,self).__init__()
        
        self.min_size = min_size
        self.max_size = max_size
        
        self.queue = Queue.Queue()
        self.stopped = False
        self.pool = []
        
        for i in xrange(max_size):
            self.pool.append(MQThread(self,self.thread_count+1))
            self.pool[-1].start()
    
    def _get_thread_count(self):
        return len(self.pool)
    thread_count = property(_get_thread_count)
    
    def process(self,message):        
        if self.stopped:
            return
            
        self.queue.put_nowait((self.invoke,message))            
        
    def get_work(self,thread):
        if self.stopped:
            raise Queue.Empty()
        
        return self.queue.get(timeout=30)
                
    def stop(self):
        self.stopped = True
        for t in self.pool:
            t.die = True

        [ t.join() for t in self.pool ]
            
class MQThread(DaemonThread):
    def __init__(self,pool,number):
        super(MQThread,self).__init__()
        self.pool = pool
        self.die = False
        self.setName('Thread #%d' % number)
        
    def run(self):
        while not self.die:
            try:
                func,message = self.pool.get_work(self)
                func(message)
            except: #Queue.Empty
                pass
        

